package com.small.nacos.common.task.engine;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.common.executor.ExecutorFactory;
import com.alibaba.nacos.common.executor.NameThreadFactory;
import com.small.nacos.common.task.AbstractDelayTask;
import com.small.nacos.common.task.NacosTaskProcessor;
import org.slf4j.Logger;

import java.util.Collection;
import java.util.HashSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Author zhoujin
 * @Date 2022/11/12 14:09
 */
public class NacosDelayTaskExecuteEngine extends AbstractNacosTaskExecuteEngine<AbstractDelayTask> {

    protected final ScheduledExecutorService processingExecutor;

    protected final ConcurrentHashMap<Object, AbstractDelayTask> tasks;

    protected final ReentrantLock lock = new ReentrantLock();

    public NacosDelayTaskExecuteEngine(String name, Logger logger) {
        this(name, 32, logger, 100L);
    }

    public NacosDelayTaskExecuteEngine(String name, int initCapacity, Logger logger, long processInterval) {
        super(logger);
        tasks = new ConcurrentHashMap<Object, AbstractDelayTask>(initCapacity);
        processingExecutor = ExecutorFactory.newSingleScheduledExecutorService(new NameThreadFactory(name));
        //延时定时任务
        processingExecutor
                .scheduleWithFixedDelay(new ProcessRunnable(this), processInterval, processInterval, TimeUnit.MILLISECONDS);
    }


    @Override
    public void addTask(Object key, AbstractDelayTask newTask) {
        AbstractDelayTask existTask = tasks.get(key);
        lock.lock();
        try {

            if (null != existTask) {
                newTask.merge(existTask);
            }
            tasks.put(key, newTask);
        } finally {
            lock.unlock();
        }
    }

    @Override
    public AbstractDelayTask removeTask(Object key) {
        AbstractDelayTask existTask = tasks.get(key);
        lock.lock();
        try {
            if (null != existTask) {
                return tasks.remove(key);
            }
            return null;
        } finally {
            lock.unlock();
        }
    }

    @Override
    public AbstractDelayTask getTask(Object key) {
        return tasks.get(key);
    }

    @Override
    public Collection<Object> getAllTaskKeys() {
        Collection<Object> keys = new HashSet<Object>();
        lock.lock();
        try {
            keys.addAll(tasks.keySet());
        } finally {
            lock.unlock();
        }
        return keys;
    }

    @Override
    public void processTasks() {
        Collection<Object> keys = getAllTaskKeys();
        for (Object taskKey : keys) {
            AbstractDelayTask task = removeTask(taskKey);
            if (null == task) {
                continue;
            }
            NacosTaskProcessor processor = getProcessor(taskKey);
            if (processor == null) {
                continue;
            }
            try {
                // 执行当前任务
                if (!processor.process(task)) {
                    //执行失败再次执行
                    retryFailedTask(taskKey, task);
                }
            } catch (Throwable e) {
                log.error("Nacos task execute error : " + e.toString(), e);
                //执行失败再次执行
                retryFailedTask(taskKey, task);
            }
        }
    }

    private void retryFailedTask(Object key, AbstractDelayTask task) {
        task.setLastProcessTime(System.currentTimeMillis());
        addTask(key, task);
    }

    @Override
    public void shutdown() throws NacosException {
        processingExecutor.shutdown();
    }

}
